#include "shmipc/com/uds_server.h"

#include <sys/socket.h>
#include <sys/types.h>
#include <sys/un.h>
#include <unistd.h>
#include <cstring>
#include <errno.h>
#include <arpa/inet.h>

#include "shmipc/com/internal/uds_com.h"
#include "shmipc/com/internal/xsocket.h"

#define UDS_ACCEPT_TIMEOUT_MS (5)
#define UDS_RECV_TIMEOUT_MS (10)

namespace shmipc::com {
using namespace shmipc::core;



UdsServer::UdsServer() {
    recv_queue_ = XMessageQueueCreate();
    recv_queue_->start();
    on_message_handler_ = std::bind(&UdsServer::onMessage, this, std::placeholders::_1);
}

UdsServer::~UdsServer() {
    unbind();
    recv_queue_ = nullptr;
}


XResultCode UdsServer::bind(const std::string& url, int32_t port /*= 0*/, OnConnectStatusHandler on_handler /*= nullptr*/) {
    if ((uds_fd_ != XBUS_INVALID_XCOM_ID) || listener_worker_ || recv_worker_) { return XResultCode::FromError(XErrC::InvalidStatus); }
    auto res = internal::XSocketServerOpen(internal::XSocketType::UDS, url, port);
    if (!res) {return XResultCode::FromError(res.error()); }

    uds_fd_ = res.value();
    is_running_ = true;
    listener_worker_ = std::make_unique<XThread>("listen:" + url, [&](){ this->onListener(); });
    recv_worker_ = std::make_unique<XThread>("recv:" + url, [&](){ this->onReceiver(); });

    return { };
}

XResultCode UdsServer::unbind() {
    if ((uds_fd_ == XBUS_INVALID_XCOM_ID) || !listener_worker_ || !recv_worker_) { return XResultCode::FromError(XErrC::InvalidStatus); }

    is_running_ = false;
    listener_worker_->join();
    listener_worker_ = nullptr;
    recv_worker_->join();
    recv_worker_ = nullptr;
    internal::XSocketClose(uds_fd_);
    uds_fd_ = XBUS_INVALID_XCOM_ID;
    recv_queue_->clear();
    return { };
}

void UdsServer::setOnMessageHandler(OnMessageHandler on_msg_handler) { /** is set this ,then 'OnMessage(XMessageSp msg);' will be overrid */
    on_message_handler_ = on_msg_handler ? on_msg_handler : std::bind(&UdsServer::onMessage, this, std::placeholders::_1);
}

XResultCode UdsServer::send(XMessageSp msg) {
    if ((uds_fd_ == XBUS_INVALID_XCOM_ID) || !listener_worker_ || !recv_worker_) { return XResultCode::FromError(XErrC::InvalidStatus); }

    auto dst_fds = msg->msg_dst();
    std::lock_guard<std::mutex> l(lock_);
    for (uint32_t i = 0; i < dst_fds.size(); i++) {
        auto it = sessions_.find(dst_fds[i]);
        if (it != sessions_.end()) {
            auto session = it->second;
            auto res = session->send(msg);
            if (!res) { //now default ignore session send error?
                //todo process error.
            }
        }
    }
    return { };
}


XResult<XMessageSp> UdsServer::allocResponse(XMessageSp request, uint32_t payload_size, uint8_t* payload /*= nullptr*/) {
    if (!request) { return XResult<XMessageSp>::FromError(XErrC::InvalidArgument); }
    XMessageType msg_type = XMessageHelper::XMsgGetResponseType(static_cast<XMessageType>(request->type()));
    if (msg_type == XMsg_TypeNone) { return XResult<XMessageSp>::FromError(XErrC::InvalidArgument);}
    auto res = XMessageAllocator::alloc(msg_type, request->message_id(), request->seq_num(), payload_size, payload);
    if (res) {
        auto msg = res.value();
        msg->add_msg_dst(request->msg_src());
        return {msg};
    }
    return res;
}

void UdsServer::onListener() {
    while (is_running_) {
        delSessions();

        auto res = internal::XSocketAccept(uds_fd_, UDS_ACCEPT_TIMEOUT_MS);

        if (res) {
            addClient(res.value());
        }
    }
}

void UdsServer::onReceiver() {
    while (is_running_) {
        auto res = recv_queue_->pop(UDS_RECV_TIMEOUT_MS);
        if (res) {
            on_message_handler_(res.value());
        }
    }
}

void UdsServer::onMessage(XMessageSp msg) {
    static_cast<void>(msg);
}

void UdsServer::addClient(XComId client_id) {
    std::lock_guard<std::mutex> l(lock_);
    auto session = std::make_shared<internal::UdsSession>(client_id, url_, [](internal::UdsSessionSp ssp){
        internal::XSocketClose(ssp->session_id());
    });

    session->setOnRecvHandler([this](XMessageSp msg){
        this->recv_queue_->push(msg);
        //todo process error.
    });
    session->setOnErrorHandler([this, client_id](XErrC errc) {
        if (errc == XErrC::Disconnected) {
            this->delClient(client_id);
        } //todo process other errors.
    });

    sessions_[client_id] = session;
}

void UdsServer::delClient(XComId client_id) {
    //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
    std::lock_guard<std::mutex> l(lock_);
    //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
    auto it = sessions_.find(client_id); 
    if (it != sessions_.end()) {
        //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
        deleted_sessions_[client_id] = it->second;
        sessions_.erase(it);
    } else {
        //printf("remove client(id:%ld) from server(%s) failed(client not found)\n", client_id, url_.c_str());
    }
}

void UdsServer::delSessions() {
    if (deleted_sessions_.size() > 0) {
        SessionMap deleted_sessions;
        {
            std::unique_lock<std::mutex> l(lock_);
            deleted_sessions = deleted_sessions_;
            deleted_sessions_.clear();
        }

        if (deleted_sessions.size() > 0) {/** release delete client */
            //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
            for (auto [client_id, session]: deleted_sessions) {
                //printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
                if (session) {
                    printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
                    auto ret = session->stop();
                    printf("[%s.%d]====>:session_id(id:0x%lx) stop ret(%s)\n", __FUNCTION__, __LINE__, 
                          session->session_id(), ret ? "success" : XErr2Str(ret.error()));
                }
            }
            printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
            deleted_sessions.clear();
            printf("[%s.%d]====>:\n", __FUNCTION__, __LINE__);
        }
    }
}

} //namespace shmipc::com